package com.xunk.equipment.websocket;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;

import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class WebsocketManager {
    //根据sid来获取对应的WebSocket
    public static ConcurrentHashMap<String, List<WebSocketServer>> WEBSOCKET_MAP = new ConcurrentHashMap<>();
    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    public static int ONLINE_COUNT = 0;

    /**
     * 点对点发送
     */
    public static void appointSending(String userId, String message) {
        if (StringUtils.isNotBlank(message) && StringUtils.isNotBlank(userId)) {
            List<WebSocketServer> servers = WEBSOCKET_MAP.get(userId);
            if (CollectionUtils.isNotEmpty(servers)) {
                log.info("推送消息到窗口  {}， 推送内容: {} , 当前存在连接数 {}", userId, message, servers.size());
                servers.forEach(server -> server.sendMessage(message));
            } else {
                log.info("意向推送到 {}, 但是没有连接存在", userId);
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return ONLINE_COUNT;
    }

    public static synchronized int addOnlineCount() {
        ONLINE_COUNT++;
        return ONLINE_COUNT;
    }

    public static synchronized int subOnlineCount() {
        ONLINE_COUNT--;
        return ONLINE_COUNT;
    }

    public static void cleanSocket() {

        log.info("开启线程定时扫描过期socket");
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        // 异步单线程 5分钟检测
        scheduler.scheduleAtFixedRate(() -> {
            WEBSOCKET_MAP.forEach((key, servers) -> servers.forEach(server -> {
                try {
                    server.cleanExpireServer();
                } catch (Exception e) {
                    log.error("清除过期socket server异常, userId = {}", key, e);
                    servers.remove(server);
                    server = null;
                }
            }));
        }, 3, 5, TimeUnit.MINUTES);
    }
}
